package io.kiki.sba.registry.server.metadata.slot.balance;


import io.kiki.sba.registry.common.model.Triple;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.slot.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.exception.SofaRegistryRuntimeException;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.server.metadata.slot.SlotBalancer;
import io.kiki.sba.registry.server.metadata.slot.util.builder.SlotsBuilder;
import io.kiki.sba.registry.server.metadata.slot.util.comparator.Comparators;
import io.kiki.sba.registry.util.MathUtils;
import lombok.Getter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

import static io.kiki.sba.registry.server.metadata.slot.balance.LeaderOnlyBalancer.TRIGGER_THESHOLD;

/**
 * 严重警告 该类比较复杂 请大家不要胡乱修改xxx$yyy$zzz链式命名，通过此命名表达依赖关系。
 *
 * @author huihui
 */
public class DefaultSlotBalancer implements SlotBalancer {

    private static final Logger logger = LoggerFactory.getLogger(DefaultSlotBalancer.class);
    @Getter
    protected final SlotsBuilder slotsBuilder;
    private final Set<String> dataStoreNodeIdSet;
    private final BalancePolicy balancePolicy = new BalancePolicyImpl();
    private final int slotNum;
    private final int slotReplicas;

    public DefaultSlotBalancer(SlotsBuilder slotsBuilder, Collection<String> dataStoreNodeIdSet) {
        this.dataStoreNodeIdSet = Collections.unmodifiableSet(new TreeSet<>(dataStoreNodeIdSet));
        this.slotsBuilder = slotsBuilder;
        this.slotNum = slotsBuilder.getSlotNums();
        this.slotReplicas = slotsBuilder.getSlotReplicas();
    }

    @Override
    public Slots balance() {
        if (dataStoreNodeIdSet.isEmpty()) {
            logger.warn("暂时没有发现数据存储节点!");
            throw new SofaRegistryRuntimeException("暂时没有发现数据存储节点!");
        }
        if (slotReplicas < TRIGGER_THESHOLD) {
            logger.warn("暂时没有配置从节点配置,[balance] slot replica[{}] means no followers, balance leader only", slotsBuilder.getSlotReplicas());
            return new LeaderOnlyBalancer(slotsBuilder, dataStoreNodeIdSet).balance();
        }
        if (balanceLeaderSlots()) {
            logger.info("[balanceLeaderSlots] end");
            slotsBuilder.incrEpoch();
            return slotsBuilder.build();
        }
        if (highWaterMarkFollowerMigrateOut()) {
            logger.info("[highWaterMarkFollowerMigrateOut] end");
            slotsBuilder.incrEpoch();
            return slotsBuilder.build();
        }
        if (lowWaterMarkFollowerMigrateIn()) {
            logger.info("[lowWaterMarkFollowerMigrateIn] end");
            slotsBuilder.incrEpoch();
            return slotsBuilder.build();
        }
        // check the low watermark leader, the follower has balanced
        // just upgrade the followers in low data server
        if (lowWaterMarkLeaderDataStoreNodeFollowerToLeaderMigrateIn()) {
            logger.info("lowWaterMarkLeaderDataStoreNodeFollowerToLeaderMigrateIn end");
            slotsBuilder.incrEpoch();
            return slotsBuilder.build();
        }
        logger.info("[balance] do nothing");
        return null;
    }

    private boolean balanceLeaderSlots() {
        // ceil avg, find the high water mark
        final int leaderAvgNumPerDataStoreNode = MathUtils.divideCeil(slotNum, dataStoreNodeIdSet.size());
        if (leastLeadersFirstDataStoreNodeChangeFollowerToLeader(leaderAvgNumPerDataStoreNode)) {
            return true;
        }
        if (highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdChangeFollowerToLeader(leaderAvgNumPerDataStoreNode)) {
            return true;
        }
        return false;
    }

    /**
     * 平滑的将负载比较的大leader由follower接替
     */
    private boolean leastLeadersFirstDataStoreNodeChangeFollowerToLeader(int leaderAvgNumPerDataStoreNode) {
        // smoothly, find the dataNode which owners the target slot's follower  and upgrade the follower to leader
        final int maxMoveLeaderSlots = balancePolicy.getMaxMoveLeaderSlots();
        final int threshold = balancePolicy.getHighWaterMarkSlotLeaderNums(leaderAvgNumPerDataStoreNode);
        int balanced = 0;
        Set<String> notSatisfiesDataStoreNodeIdSet = new HashSet<>();

        while (balanced < maxMoveLeaderSlots) {
            int last = balanced;
            // 1. find the dataNode which has leaders more than high water mark and sorted by leaders.num desc
            final List<String> highWaterMarkLeaderDataStoreNodeIdList = getMostLeadersFirstDataStoreNodeIdListByLeaderHighWaterMark(threshold);
            if (highWaterMarkLeaderDataStoreNodeIdList.isEmpty()) {
                break;
            }
            // could not found any follower to upgrade
            if (notSatisfiesDataStoreNodeIdSet.containsAll(highWaterMarkLeaderDataStoreNodeIdList)) {
                logger.info("[upgradeHighLeaders] could not found followers to upgrade for {}", highWaterMarkLeaderDataStoreNodeIdList);
                break;
            }
            // 2. find the dataNode which could own a new leader exclude the high
            final Set<String> excludesDataStoreNodeIdSet = new HashSet<>(highWaterMarkLeaderDataStoreNodeIdList);
            // exclude the dataNode which could not add any leader
            excludesDataStoreNodeIdSet.addAll(getMostLeadersFirstDataStoreNodeIdListByLeaderHighWaterMark(threshold - 1));
            //
            for (String highWaterMarkLeaderDataStoreNodeId : highWaterMarkLeaderDataStoreNodeIdList) {
                if (notSatisfiesDataStoreNodeIdSet.contains(highWaterMarkLeaderDataStoreNodeId)) {
                    continue;
                }
                Tuple<String, Integer> tuple = leastLeadersFirstDataStoreNodeChangeFollowerToLeader(highWaterMarkLeaderDataStoreNodeId, excludesDataStoreNodeIdSet);
                if (tuple == null) {
                    notSatisfiesDataStoreNodeIdSet.add(highWaterMarkLeaderDataStoreNodeId);
                    continue;
                }
                final int slotId = tuple.o2;
                final String newLeaderDataStoreNodeId = tuple.o1;
                slotsBuilder.replaceLeaderDataStoreNodeId(slotId, newLeaderDataStoreNodeId);
                //
                logger.info("[upgradeHighLeaders] slotId={} leader balance from {} to {}", slotId, highWaterMarkLeaderDataStoreNodeId, newLeaderDataStoreNodeId);
                Metrics.SlotBalance.onLeaderUpgrade(highWaterMarkLeaderDataStoreNodeId, newLeaderDataStoreNodeId, slotId);
                balanced++;
            }
            if (last == balanced)
                break;
        }
        return balanced != 0;
    }

    private Tuple<String, Integer> leastLeadersFirstDataStoreNodeChangeFollowerToLeader(String highWaterMarkLeaderDataStoreNodeId, Set<String> excludes) {
        final DataStoreNode highWaterMarkLeaderDataStoreNode = slotsBuilder.getDataStoreNode(highWaterMarkLeaderDataStoreNodeId);
        Set<Integer> highWaterMarkLeaderDataStoreNode$leaderSlotIdSet = highWaterMarkLeaderDataStoreNode.getLeaderSlotIdSet();
        //
        Map<String, List<Integer>> highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap = new HashMap<>();
        //
        for (int highWaterMarkLeaderDataStoreNode$leaderSlotId : highWaterMarkLeaderDataStoreNode$leaderSlotIdSet) {
            List<String> highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdList = slotsBuilder.getFollowerDataStoreNodeIdList(highWaterMarkLeaderDataStoreNode$leaderSlotId);
            highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdList = getCandidateDataStoreNodeIdList(highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdList, excludes, null);
            for (String highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeId : highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdList) {
                List<Integer> _slotIdList_ = highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap.computeIfAbsent(highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeId, k -> new ArrayList<>());
                _slotIdList_.add(highWaterMarkLeaderDataStoreNode$leaderSlotId);
            }
        }
        if (highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap.isEmpty()) {
            logger.info("[LeaderUpgradeOut] {} owns leader slots={}, no dataServers could be upgrade, slotId={}", highWaterMarkLeaderDataStoreNodeId, highWaterMarkLeaderDataStoreNode$leaderSlotIdSet.size(), highWaterMarkLeaderDataStoreNode$leaderSlotIdSet);
            return null;
        } else {
            logger.info("[LeaderUpgradeOut] {} owns leader slots={}, slotIds={}, upgrade candidates {}", highWaterMarkLeaderDataStoreNodeId, highWaterMarkLeaderDataStoreNode$leaderSlotIdSet.size(), highWaterMarkLeaderDataStoreNode$leaderSlotIdSet, highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap);
        }
        // sort the dataServer by leaders.num asc
        List<String> candidateDataStoreNodeIdSet = new ArrayList<>(highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap.keySet());
        /**
         * leader最少的节点排序
         */
        candidateDataStoreNodeIdSet.sort(Comparators.leastLeadersFirst(slotsBuilder));
        final String selectedDataStoreNodeId = candidateDataStoreNodeIdSet.get(0);
        List<Integer> _slotIdList_ = highWaterMarkLeaderDataStoreNode$leaderSlotId$candidateFollowerDataStoreNodeIdToFollowerSlotIdListMap.get(selectedDataStoreNodeId);
        return Tuple.of(selectedDataStoreNodeId, _slotIdList_.get(0));//优先取第一个
    }

    private boolean highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdChangeFollowerToLeader(int leaderAvgNumPerDataServer) {
        // could not found the follower to upgrade, migrate follower first
        final int maxMoveFollowerSlots = balancePolicy.getMaxMoveFollowerSlots();
        final int threshold = balancePolicy.getHighWaterMarkSlotLeaderNums(leaderAvgNumPerDataServer);

        int balanced = 0;
        while (balanced < maxMoveFollowerSlots) {
            int last = balanced;
            // 1. find the dataNode which has leaders more than high water mark   and sorted by leaders.num desc
            final List<String> highWaterMarkLeaderDataStoreNodeIdList = getMostLeadersFirstDataStoreNodeIdListByLeaderHighWaterMark(threshold);
            if (highWaterMarkLeaderDataStoreNodeIdList.isEmpty()) {
                return false;
            }
            // 2. find the dataNode which could own a new leader exclude the high
            final Set<String> excludesDataStoreNodeIdSet = new HashSet<>(highWaterMarkLeaderDataStoreNodeIdList);
            // exclude the dataNode which could not add any leader
            excludesDataStoreNodeIdSet.addAll(getMostLeadersFirstDataStoreNodeIdListByLeaderHighWaterMark(threshold - 1));
            final Set<String> newFollowerDataStoreNodeIdSet = new HashSet<>();
            // only balance highDataServer once at one round, avoid the follower moves multi times
            for (String highWaterMarkLeaderDataStoreNodeId : highWaterMarkLeaderDataStoreNodeIdList) {
                Triple<String, Integer, String> triple = highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdChangeFollowerToLeader(highWaterMarkLeaderDataStoreNodeId, excludesDataStoreNodeIdSet, newFollowerDataStoreNodeIdSet);
                if (triple == null) {
                    logger.warn("[migrateHighLeaders] could not find dataServer to migrate follower for {}", highWaterMarkLeaderDataStoreNodeId);
                    continue;
                }
                final String oldFollowerDataStoreNodeId = triple.getFirst();
                final int slotId = triple.getMiddle();
                final String newFollowerDataStoreNodeId = triple.getLast();
                slotsBuilder.removeFollowerDataStoreNodeId(slotId, oldFollowerDataStoreNodeId);
                slotsBuilder.addFollowerDataStoreNodeId(slotId, newFollowerDataStoreNodeId);
                newFollowerDataStoreNodeIdSet.add(newFollowerDataStoreNodeId);
                logger.info("[migrateHighLeaders] slotId={}, follower balance from {} to {}", slotId, oldFollowerDataStoreNodeId, newFollowerDataStoreNodeId);
                Metrics.SlotBalance.onLeaderMigrate(oldFollowerDataStoreNodeId, newFollowerDataStoreNodeId, slotId);
                balanced++;
            }
            if (last == balanced)
                break;
        }
        return balanced != 0;
    }


    private Triple<String, Integer, String> highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdChangeFollowerToLeader(String highWaterMarkLeaderDataStoreNodeId, Set<String> excludesDataStoreNodeIdSet, Set<String> newFollowerDataServers) {
        final DataStoreNode highWaterMarkLeaderDataStore = slotsBuilder.getDataStoreNode(highWaterMarkLeaderDataStoreNodeId);
        Set<Integer> highWaterMarkLeaderDataStore$leaderSlotIdSet = highWaterMarkLeaderDataStore.getLeaderSlotIdSet();
        /**
         * 新的承载节点关系
         */
        Map<String, List<Integer>> highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdToFollowerSlotIdListMap = new HashMap<>();
        for (int highWaterMarkLeaderDataStore$leaderSlotId : highWaterMarkLeaderDataStore$leaderSlotIdSet) {
            List<String> highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdList = slotsBuilder.getFollowerDataStoreNodeIdList(highWaterMarkLeaderDataStore$leaderSlotId);
            for (String highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId : highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdList) {
                if (newFollowerDataServers.contains(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId)) {
                    // the followerDataServer contains move in follower, could not be move out candidates
                    continue;
                }
                List<Integer> followerSlotIdList = highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdToFollowerSlotIdListMap.computeIfAbsent(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId, k -> new ArrayList<>());
                followerSlotIdList.add(highWaterMarkLeaderDataStore$leaderSlotId);
            }
        }
        logger.info("[selectFollower4LeaderMigrate] {} owns leader slots={}, slotIds={}, migrate candidates {}, newFollowers={}", highWaterMarkLeaderDataStoreNodeId, highWaterMarkLeaderDataStore$leaderSlotIdSet.size(), highWaterMarkLeaderDataStore$leaderSlotIdSet, highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdToFollowerSlotIdListMap, newFollowerDataServers);
        // sort the dataServer by follower.num desc
        List<String> highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdList = new ArrayList<>(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdToFollowerSlotIdListMap.keySet());
        highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdList.sort(Comparators.mostFollowersFirst(slotsBuilder));
        for (String highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId : highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdList) {
            final List<Integer> followerSlotIdList = highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeIdToFollowerSlotIdListMap.get(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId);
            for (Integer followerSlotId : followerSlotIdList) {
                // chose the dataServer which own least leaders
                /**
                 * leastLeadersFirst
                 */
                List<String> candidateDataStoreNodeIdSet = getCandidateDataStoreNodeIdList(dataStoreNodeIdSet, excludesDataStoreNodeIdSet, Comparators.leastLeadersFirst(slotsBuilder));
                for (String candidateDataStoreNodeId : candidateDataStoreNodeIdSet) {
                    if (candidateDataStoreNodeId.equals(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId)) {
                        // the same, skip
                        continue;
                    }
                    DataStoreNode candidateDataStoreNode = slotsBuilder.getDataStoreNode(candidateDataStoreNodeId);
                    if (candidateDataStoreNode.containsFollowerSlotId(followerSlotId)) {
                        logger.error("[selectFollower4LeaderMigrate] slotId={}, follower conflict with migrate from {} to {}", followerSlotId, highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId, candidateDataStoreNode);
                        continue;
                    }
                    return Triple.from(highWaterMarkLeaderDataStore$leaderSlotId$followerDataStoreNodeId, followerSlotId, candidateDataStoreNodeId);
                }
            }
        }
        return null;
    }


    /**
     * 高负载从节点分配
     *
     * @return
     */
    private boolean highWaterMarkFollowerMigrateOut() {
        final int followerAvgCount = MathUtils.divideCeil(slotNum * (slotReplicas - 1), dataStoreNodeIdSet.size());
        final int threshold = balancePolicy.getHighWaterMarkSlotFollowerNums(followerAvgCount);
        //
        int balanced = 0, prevBalanced = -1;
        final int maxMoveFollowerSlots = balancePolicy.getMaxMoveFollowerSlots();
        while (balanced < maxMoveFollowerSlots) {
            final List<String> highWaterMarkFollowerDataStoreNodeIdList = getMostFollowersFirstDataStoreNodeIdListByFollowerHighWaterMark(threshold);
            if (highWaterMarkFollowerDataStoreNodeIdList.isEmpty()) {
                break;
            }
            Set<String> excludesDataStoreNodeIdList = new HashSet<>(highWaterMarkFollowerDataStoreNodeIdList);
            excludesDataStoreNodeIdList.addAll(getMostFollowersFirstDataStoreNodeIdListByFollowerHighWaterMark(threshold - 1));
            //
            prevBalanced = balanced;
            for (String highWaterMarkFollowerDataStoreNodeId : highWaterMarkFollowerDataStoreNodeIdList) {
                Tuple<String, Integer> tuple = highWaterMarkFollowerMigrateOut(highWaterMarkFollowerDataStoreNodeId, excludesDataStoreNodeIdList);
                if (tuple == null) {
                    logger.warn("[balanceHighFollowerSlots] could not find follower slot to balance: {}", highWaterMarkFollowerDataStoreNodeId);
                    continue;
                }
                final int followerSlotId = tuple.o2;
                final String newFollowerDataStoreNodeId = tuple.o1;
                //
                slotsBuilder.removeFollowerDataStoreNodeId(followerSlotId, highWaterMarkFollowerDataStoreNodeId);
                slotsBuilder.addFollowerDataStoreNodeId(followerSlotId, newFollowerDataStoreNodeId);
                logger.info("[balanceHighFollowerSlots] balance follower slotId={} from {} to {}", followerSlotId, highWaterMarkFollowerDataStoreNodeId, newFollowerDataStoreNodeId);
                Metrics.SlotBalance.onHighFollowerMigrate(highWaterMarkFollowerDataStoreNodeId, newFollowerDataStoreNodeId, followerSlotId);
                balanced++;
                break;
            }
            /**
             * avoid for infinity loop once the "prev-balanced == balanced", it means that we can't find
             * any suitable candidate for migrate stop before we case an infinity loop
             */
            if (prevBalanced == balanced) {
                logger.warn("[balanceHighFollowerSlots][prevBlanced == balanced]no more balance available");
                break;
            }
        }
        return balanced != 0;
    }

    /**
     * 高负载follower节点迁出
     */
    private Tuple<String, Integer> highWaterMarkFollowerMigrateOut(String highWaterMarkFollowerDataStoreNodeId, Set<String> excludesDataStoreNodeIdList) {
        final DataStoreNode highWaterMarkFollowerDataStoreNode = slotsBuilder.getDataStoreNode(highWaterMarkFollowerDataStoreNodeId);
        Set<Integer> highWaterMarkFollowerDataStoreNode$followerSlotIdSet = highWaterMarkFollowerDataStoreNode.getFollowerSlotIdSet();
        /**
         * follower低负载排序
         */
        List<String> candidateDataStoreNodeIdSet =//
                getCandidateDataStoreNodeIdList(dataStoreNodeIdSet,//
                        excludesDataStoreNodeIdList,//
                        Comparators.leastFollowersFirst(slotsBuilder));//
        logger.info("[selectFollower4BalanceOut] target={}, followerSize={}, candidates={}", highWaterMarkFollowerDataStoreNodeId, highWaterMarkFollowerDataStoreNode$followerSlotIdSet.size(), candidateDataStoreNodeIdSet);
        for (Integer highWaterMarkFollowerDataStoreNode$followerSlotId : highWaterMarkFollowerDataStoreNode$followerSlotIdSet) {
            for (String candidateDataStoreNodeId : candidateDataStoreNodeIdSet) {
                DataStoreNode candidateDataStoreNode = slotsBuilder.getDataStoreNode(candidateDataStoreNodeId);
                if (candidateDataStoreNode.containsLeaderSlotId(highWaterMarkFollowerDataStoreNode$followerSlotId)) {
                    logger.info("[selectFollower4BalanceOut] skip, conflict leader, target={}, follower={}, candidate={}", highWaterMarkFollowerDataStoreNodeId, highWaterMarkFollowerDataStoreNode$followerSlotId, candidateDataStoreNodeId);
                    continue;
                }
                if (candidateDataStoreNode.containsFollowerSlotId(highWaterMarkFollowerDataStoreNode$followerSlotId)) {
                    logger.info("[selectFollower4BalanceOut] skip, conflict follower, target={}, follower={}, candidate={}", highWaterMarkFollowerDataStoreNodeId, highWaterMarkFollowerDataStoreNode$followerSlotId, candidateDataStoreNodeId);
                    continue;
                }
                return Tuple.of(candidateDataStoreNodeId, highWaterMarkFollowerDataStoreNode$followerSlotId);
            }
        }
        return null;
    }

    /**
     * 把其他高负载的follow迁移到低负载的数据存储节点
     *
     * @return
     */
    private boolean lowWaterMarkFollowerMigrateIn() {
        final int followerFloorAvg = Math.floorDiv(slotNum * (slotReplicas - 1), dataStoreNodeIdSet.size());
        final int maxMoveFollowerSlots = balancePolicy.getMaxMoveFollowerSlots();
        final int threshold = balancePolicy.getLowWaterMarkSlotFollowerNums(followerFloorAvg);
        int balanced = 0, prevBalanced = -1;
        while (balanced < maxMoveFollowerSlots) {
            final List<String> lowWaterMarkFollowerDataStoreNodeIdList = getLeastFollowersFirstDataStoreNodeIdListByFollowerLowWaterMark(threshold);
            if (lowWaterMarkFollowerDataStoreNodeIdList.isEmpty()) {
                break;
            }

            Set<String> excludesDataStoreNodeIdList = new HashSet<>(lowWaterMarkFollowerDataStoreNodeIdList);
            excludesDataStoreNodeIdList.addAll(getLeastFollowersFirstDataStoreNodeIdListByFollowerLowWaterMark(threshold + 1));

            prevBalanced = balanced;
            for (String lowWaterMarkFollowerDataStoreNodeId : lowWaterMarkFollowerDataStoreNodeIdList) {
                Tuple<String, Integer> tuple = lowWaterMarkFollowerMigrateIn(lowWaterMarkFollowerDataStoreNodeId, excludesDataStoreNodeIdList);
                if (tuple == null) {
                    logger.warn("[balanceLowFollowerSlots] could not find follower slot to balance: {}", lowWaterMarkFollowerDataStoreNodeId);
                    continue;
                }
                final int followerSlotId = tuple.o2;
                final String oldFollowerDataStoreNodeId = tuple.o1;
                slotsBuilder.removeFollowerDataStoreNodeId(followerSlotId, oldFollowerDataStoreNodeId);
                slotsBuilder.addFollowerDataStoreNodeId(followerSlotId, lowWaterMarkFollowerDataStoreNodeId);
                logger.info("[balanceLowFollowerSlots] balance follower slotId={} from {} to {}", followerSlotId, oldFollowerDataStoreNodeId, lowWaterMarkFollowerDataStoreNodeId);
                Metrics.SlotBalance.onLowFollowerMigrate(oldFollowerDataStoreNodeId, lowWaterMarkFollowerDataStoreNodeId, followerSlotId);
                balanced++;
                break;
            }
            if (prevBalanced == balanced) {
                logger.warn("[balanceLowFollowerSlots][prevBlanced == balanced]no more balance available");
                break;
            }
        }
        return balanced != 0;
    }

    /**
     * 低负载follower节点迁入
     */
    private Tuple<String, Integer> lowWaterMarkFollowerMigrateIn(String lowWaterMarkFollowerDataStoreNodeId, Set<String> excludesDataStoreNodeIdList) {
        final DataStoreNode lowWaterMarkFollowerDataStoreNode = slotsBuilder.getDataStoreNode(lowWaterMarkFollowerDataStoreNodeId);
        /**
         * 高负载优先
         */
        List<String> candidateDataStoreNodeIdSet = getCandidateDataStoreNodeIdList(//
                dataStoreNodeIdSet,//
                excludesDataStoreNodeIdList,//
                Comparators.mostFollowersFirst(slotsBuilder));//
        logger.info("[selectFollower4BalanceIn] target={}, followerSize={}, candidates={}", lowWaterMarkFollowerDataStoreNodeId, lowWaterMarkFollowerDataStoreNode.getFollowerSlotIdSet().size(), candidateDataStoreNodeIdSet);
        for (String candidateDataStoreNodeId : candidateDataStoreNodeIdSet) {
            DataStoreNode candidateDataStoreNode = slotsBuilder.getDataStoreNode(candidateDataStoreNodeId);
            /**
             * 暂时只迁移从
             */
            Set<Integer> candidateDataStoreNode$followerSlotIdSet = candidateDataStoreNode.getFollowerSlotIdSet();
            for (int candidateDataStoreNode$followerSlotId : candidateDataStoreNode$followerSlotIdSet) {
                if (lowWaterMarkFollowerDataStoreNode.containsFollowerSlotId(candidateDataStoreNode$followerSlotId)) {
                    logger.info("[selectFollower4BalanceIn] skip, target {} contains follower {}, candidate={}", lowWaterMarkFollowerDataStoreNodeId, candidateDataStoreNode$followerSlotId, candidateDataStoreNodeId);
                    continue;
                }
                if (lowWaterMarkFollowerDataStoreNode.containsLeaderSlotId(candidateDataStoreNode$followerSlotId)) {
                    logger.info("[selectFollower4BalanceIn] skip, target {} contains leader {}, candidate={}", lowWaterMarkFollowerDataStoreNodeId, candidateDataStoreNode$followerSlotId, candidateDataStoreNodeId);
                    continue;
                }
                return Tuple.of(candidateDataStoreNodeId, candidateDataStoreNode$followerSlotId);
            }
        }
        return null;
    }

    /**
     * 如果节点leader比较少则进行分配
     *
     * @return
     */
    private boolean lowWaterMarkLeaderDataStoreNodeFollowerToLeaderMigrateIn() {
        final int leaderFloorAvgCount = Math.floorDiv(slotNum, dataStoreNodeIdSet.size());
        final int maxMoveLeaderSlots = balancePolicy.getMaxMoveLeaderSlots();
        final int threshold = balancePolicy.getLowWaterMarkSlotLeaderNums(leaderFloorAvgCount);
        int balanced = 0;
        Set<String> notSatisfiesDataStoreNodeIdSet = new HashSet<>();

        while (balanced < maxMoveLeaderSlots) {
            int last = balanced;
            // 1. find the dataNode which has leaders less than low water mark  and sorted by leaders.num asc
            final List<String> lowWaterMarkLeaderDataStoreNodeIdList = getLeastLeadersFirstDataStoreNodeIdListByLeaderLowWaterMark(threshold);
            if (lowWaterMarkLeaderDataStoreNodeIdList.isEmpty()) {
                break;
            }
            // could not found any follower to upgrade
            if (notSatisfiesDataStoreNodeIdSet.containsAll(lowWaterMarkLeaderDataStoreNodeIdList)) {
                logger.info("[upgradeLowLeaders] could not found followers to upgrade for {}", lowWaterMarkLeaderDataStoreNodeIdList);
                break;
            }
            // 2. find the dataNode which could not remove a leader exclude the low
            final Set<String> excludesDataStoreNodeIdSet = new HashSet<>(lowWaterMarkLeaderDataStoreNodeIdList);
            // exclude the dataNode which could not remove any leader
            excludesDataStoreNodeIdSet.addAll(getLeastLeadersFirstDataStoreNodeIdListByLeaderLowWaterMark(threshold + 1));
            for (String lowWaterMarkLeaderDataStoreNodeId : lowWaterMarkLeaderDataStoreNodeIdList) {
                if (notSatisfiesDataStoreNodeIdSet.contains(lowWaterMarkLeaderDataStoreNodeId)) {
                    continue;
                }
                Tuple<String, Integer> tuple = lowWaterMarkLeaderDataStoreNodeFollowerToLeaderMigrateIn(lowWaterMarkLeaderDataStoreNodeId, excludesDataStoreNodeIdSet);
                if (tuple == null) {
                    notSatisfiesDataStoreNodeIdSet.add(lowWaterMarkLeaderDataStoreNodeId);
                    continue;
                }
                final String oldLeaderDataStoreNodeId = tuple.o1;
                final int slotId = tuple.o2;
                //
                final String replaceLeaderDataStoreNodeId = slotsBuilder.replaceLeaderDataStoreNodeId(slotId, lowWaterMarkLeaderDataStoreNodeId);//变成自己
                if (!StringUtils.equals(oldLeaderDataStoreNodeId, replaceLeaderDataStoreNodeId)) {
                    logger.error("[upgradeLowLeaders] conflict leader, slotId={} leader balance from {}/{} to {}", slotId, oldLeaderDataStoreNodeId, replaceLeaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNodeId);
                    throw new SofaRegistryRuntimeException(String.format("upgradeLowLeaders, conflict leader=%d of %s and %s", slotId, oldLeaderDataStoreNodeId, replaceLeaderDataStoreNodeId));
                }
                Metrics.SlotBalance.onLowLeaderReplace(oldLeaderDataStoreNodeId, replaceLeaderDataStoreNodeId, slotId);
                logger.info("[upgradeLowLeaders] slotId={} leader balance from {} to {}", slotId, oldLeaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNodeId);
                balanced++;
            }
            if (last == balanced)
                break;
        }
        return balanced != 0;
    }

    private Tuple<String, Integer> lowWaterMarkLeaderDataStoreNodeFollowerToLeaderMigrateIn(String lowWaterMarkLeaderDataStoreNodeId, Set<String> excludesDataStoreNodeIdSet) {
        final DataStoreNode lowWaterMarkLeaderDataStoreNode = slotsBuilder.getDataStoreNode(lowWaterMarkLeaderDataStoreNodeId);
        Set<Integer> lowWaterMarkLeaderDataStoreNode$followerSlotIdSet = lowWaterMarkLeaderDataStoreNode.getFollowerSlotIdSet();
        logger.info("[LeaderUpgradeIn] {} find follower to upgrade, {}/{}", lowWaterMarkLeaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNode$followerSlotIdSet.size(), lowWaterMarkLeaderDataStoreNode$followerSlotIdSet);
        Map<String, List<Integer>> lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap = new HashMap<>();
        for (int lowWaterMarkLeaderDataStoreNode$followerSlotId : lowWaterMarkLeaderDataStoreNode$followerSlotIdSet) {
            final String lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId = slotsBuilder.getLeaderDataStoreNodeId(lowWaterMarkLeaderDataStoreNode$followerSlotId);
            if (StringUtils.isBlank(lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId)) {
                // no leader, should not happen
                logger.error("[LeaderUpgradeIn] no leader for slotId={} in {}", lowWaterMarkLeaderDataStoreNode$followerSlotId, lowWaterMarkLeaderDataStoreNodeId);
                continue;
            }
            /**
             * 已经很闲了 就不用管了
             */
            if (excludesDataStoreNodeIdSet.contains(lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId)) {
                final DataStoreNode leaderDataStoreNode = slotsBuilder.getDataStoreNode(lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId);
                logger.info("[LeaderUpgradeIn] {} not owns enough leader to downgrade slotId={} for {}, leaderSize={}", lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNode$followerSlotId, lowWaterMarkLeaderDataStoreNodeId, leaderDataStoreNode.getLeaderSlotIdSet().size());
                continue;
            }

            List<Integer> leaderSlotIdList = lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap.computeIfAbsent(lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeId, k -> new ArrayList<>());
            leaderSlotIdList.add(lowWaterMarkLeaderDataStoreNode$followerSlotId);
        }
        if (lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap.isEmpty()) {
            logger.info("[LeaderUpgradeIn] {} owns followerSize={}, no dataServers could be downgrade, slotId={}", lowWaterMarkLeaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNode$followerSlotIdSet.size(), lowWaterMarkLeaderDataStoreNode$followerSlotIdSet);
            return null;
        } else {
            logger.info("[LeaderUpgradeIn] {} owns followerSize={}, slotIds={}, downgrade candidates {}", lowWaterMarkLeaderDataStoreNodeId, lowWaterMarkLeaderDataStoreNode$followerSlotIdSet.size(), lowWaterMarkLeaderDataStoreNode$followerSlotIdSet, lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap);
        }
        // sort the dataServer by leaders.num asc
        List<String> lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdList = new ArrayList<>(lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap.keySet());
        /**
         * 负载最小的leader
         */
        lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdList.sort(Comparators.mostLeadersFirst(slotsBuilder));
        final String selectedLeaderDataStoreNodeId = lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdList.get(0);
        List<Integer> selectedLeaderSlotIdList = lowWaterMarkLeaderDataStoreNode$followerSlotId$leaderDataStoreNodeIdToLeaderSlotIdListMap.get(selectedLeaderDataStoreNodeId);
        return Tuple.of(selectedLeaderDataStoreNodeId, selectedLeaderSlotIdList.get(0));//取值第一个,暂时没有发现任何依据
    }


    private List<String> getLeastLeadersFirstDataStoreNodeIdListByLeaderLowWaterMark(int threshold) {
        List<DataStoreNode> dataStoreNodeList = slotsBuilder.getDataNodeSlotsLeaderBelow(threshold);
        List<String> dataStoreNodeIdList = DataStoreNode.collectDataStoreNodeIdList(dataStoreNodeList);
        dataStoreNodeIdList.sort(Comparators.leastLeadersFirst(slotsBuilder));
        logger.info("[LeaderLowWaterMark] threshold={}, dataServers={}/{}", threshold, dataStoreNodeIdList.size(), dataStoreNodeIdList);
        return dataStoreNodeIdList;
    }

    private List<String> getMostLeadersFirstDataStoreNodeIdListByLeaderHighWaterMark(int threshold) {
        List<DataStoreNode> dataStoreNodes = slotsBuilder.getDataNodeSlotsLeaderBeyond(threshold);
        List<String> dataStoreNodeIdList = DataStoreNode.collectDataStoreNodeIdList(dataStoreNodes);
        dataStoreNodeIdList.sort(Comparators.mostLeadersFirst(slotsBuilder));
        logger.info("[LeaderHighWaterMark] threshold={}, dataServers={}/{}", threshold, dataStoreNodeIdList.size(), dataStoreNodeIdList);
        return dataStoreNodeIdList;
    }

    private List<String> getLeastFollowersFirstDataStoreNodeIdListByFollowerLowWaterMark(int threshold) {
        List<DataStoreNode> dataStoreNodeList = slotsBuilder.getDataNodeSlotsFollowerBelow(threshold);
        List<String> dataStoreNodeIdList = DataStoreNode.collectDataStoreNodeIdList(dataStoreNodeList);
        dataStoreNodeIdList.sort(Comparators.leastFollowersFirst(slotsBuilder));
        logger.info("[FollowerLowWaterMark] threshold={}, dataServers={}/{}", threshold, dataStoreNodeIdList.size(), dataStoreNodeIdList);
        return dataStoreNodeIdList;
    }


    private List<String> getMostFollowersFirstDataStoreNodeIdListByFollowerHighWaterMark(int threshold) {
        List<DataStoreNode> dataStoreNodeList = slotsBuilder.getDataNodeSlotsFollowerBeyond(threshold);
        List<String> dataStoreNodeIdList = DataStoreNode.collectDataStoreNodeIdList(dataStoreNodeList);
        dataStoreNodeIdList.sort(Comparators.mostFollowersFirst(slotsBuilder));
        logger.info("[FollowerHighWaterMark] threshold={}, dataServers={}/{}", threshold, dataStoreNodeIdList.size(), dataStoreNodeIdList);
        return dataStoreNodeIdList;
    }

    private List<String> getCandidateDataStoreNodeIdList(Collection<String> candidateDataStoreNodeIdList, Collection<String> excludesDataStoreNodeIdList, Comparator<String> comparator) {
        Set<String> candidateDataStoreNodeIdSet = new HashSet<>(candidateDataStoreNodeIdList);
        candidateDataStoreNodeIdSet.removeAll(excludesDataStoreNodeIdList);
        List<String> list = new ArrayList<>(candidateDataStoreNodeIdSet);
        if (comparator != null) {
            list.sort(comparator);
        }
        return list;
    }
}
